package com.yonyou.common.service.messagecenter.rabbitmq;

import java.io.IOException;
import java.sql.Date;
import java.util.concurrent.TimeoutException;

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.rabbitmq.client.MessageProperties;
import com.yonyou.common.service.messagecenter.dao.TtMessageHistoryMapper;
import com.yonyou.common.service.messagecenter.rabbitmq.entity.MessageProtocol;
import com.yonyou.common.service.messagecenter.rabbitmq.entity.TtMessageHistory;
import com.yonyou.common.service.messagecenter.rabbitmq.service.MessageHistoryService;
import com.yonyou.common.service.messagecenter.rabbitmq.service.MessageHistorySyncService;
import com.yonyou.common.service.messagecenter.rabbitmq.service.MessageProtocolService;
import com.yonyou.common.service.messagecenter.util.MqConsts;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;

@Service
public class MessageProducer extends EndPoint{

	private static Logger logger = LoggerFactory.getLogger(MessageProducer.class);
	@Autowired
	private MessageHistoryService messageHistoryService;
	@Autowired
	private MessageProtocolService messageProtocolService;
	@Autowired
	private MessageHistorySyncService messageHistorySyncService;
	private volatile MessageConverter jackson2JsonMessageConverter=new Jackson2JsonMessageConverter();
	private volatile MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();
	public MessageProducer() throws IOException, TimeoutException {
		super();
	}

//	public synchronized boolean sendMessageSimple(String exchange,String routKey,MessageProtocol messageProtocol,boolean reSend) {
//		Channel channel=this.getProduceChannel();
//    	try{
//    		logger.info("-----id:"+Thread.currentThread().getId());
//        	org.springframework.amqp.core.MessageProperties prop=
//        			new org.springframework.amqp.core.MessageProperties();
//        	prop.setContentType(org.springframework.amqp.core.MessageProperties.CONTENT_TYPE_JSON);
//    		String data=messageProtocolService.formatProduceData(messageProtocol);
//    		channel.txSelect();
//    		Message message=jackson2JsonMessageConverter.toMessage(data, prop);
//    		org.springframework.amqp.core.MessageProperties messageProperties = message.getMessageProperties();
//    		BasicProperties convertedMessageProperties = this.messagePropertiesConverter
//    				.fromMessageProperties(messageProperties, "UTF-8");
//    		channel.basicPublish(exchange, routKey, false, convertedMessageProperties, data.getBytes());//message.getBody()
//    		channel.txCommit();
//    	}catch(Exception ex){
//    		logger.info("--error:"+ex.getMessage());
//			try {
//				channel.txRollback();
//			} catch (IOException e) {
//				//e.printStackTrace();
//			    logger.info("发生异常："+e.getMessage());
//			}
//    		//ex.printStackTrace();
//            logger.info("发生异常："+ex.getMessage());
//    		return false;
//    	}
//        return true;
//    }
    public synchronized boolean reSendMessage(String exchange,String routKey,MessageProtocol messageProtocol,boolean reSend) {
    	return send(exchange,routKey,messageProtocol,reSend);
    }
    public  boolean send(String exchange,String routKey,MessageProtocol messageProtocol,boolean reSend) {
    	TtMessageHistory record=null;
		Channel channel=this.getProduceChannel();
		if(channel==null){
            logger.info("---无法取得MQ连接");
    		return false;
		}
    	try{
    		logger.info("-----id:"+Thread.currentThread().getId());
    		//MessageProperties.PERSISTENT_TEXT_PLAIN;
        	org.springframework.amqp.core.MessageProperties prop=
        			new org.springframework.amqp.core.MessageProperties();
        	prop.setContentType(org.springframework.amqp.core.MessageProperties.CONTENT_TYPE_JSON);
    		String data=messageProtocolService.formatProduceData(messageProtocol);
//    		logger.info("-----queue data:"+data);
    		record=messageHistoryService.getMessageHistory(messageProtocol,reSend,exchange);
    		//开启事务模式
    		channel.txSelect();
    		Message message=jackson2JsonMessageConverter.toMessage(data, prop);
    		org.springframework.amqp.core.MessageProperties messageProperties = message.getMessageProperties();
    		BasicProperties convertedMessageProperties = this.messagePropertiesConverter
    				.fromMessageProperties(messageProperties, "UTF-8");
    		//this.getChannel().basicPublish(exchange,routKey, MessageProperties.PERSISTENT_TEXT_PLAIN, tmp.getBytes());//SerializationUtils.serialize(object)
//    		messageHistoryService.insertHistory(record);
    		channel.basicPublish(exchange, routKey, false, convertedMessageProperties, data.getBytes());//message.getBody()
    		channel.txCommit();
    		try{
        		messageHistorySyncService.insertHistory(record);
    		}catch(Exception e){
        		logger.error("---error:"+e.getMessage());
    		}
//    		messageHistoryService.updateHistorySuccess(record);
    	}catch(Exception ex){
    		logger.error("---error:"+ex.getMessage());
			try {
				channel.txRollback();
			} catch (IOException e) {
				//e.printStackTrace();
	            logger.info("发生异常："+e.getMessage());
			}
//    		messageHistoryService.updateHistoryFail(record);
//			record.setStatus(MqConsts.PRODUCE_STATUS_FAIL);
//			messageHistorySyncService.insertHistory(record);
    		//ex.printStackTrace();
            logger.info("发生异常："+ex.getMessage());
    		return false;
    	}
        return true;
    }

    public  boolean reqSendMessage(String exchange,String routKey,MessageProtocol messageProtocol) {
    	boolean result=false;
		result=send(exchange,routKey,messageProtocol,false);
		return result;
    	 
    }
}
